package com.atguigu.gamll.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gamll.realtime.app.dwd.BaseApp;
import com.atguigu.gamll.realtime.app.func.DimAsyncFunction;
import com.atguigu.gamll.realtime.beans.OrderWide;
import com.atguigu.gamll.realtime.beans.PaymentWide;
import com.atguigu.gamll.realtime.beans.ProductStats;
import com.atguigu.gamll.realtime.beans.VisitorStats;
import com.atguigu.gamll.realtime.common.GmallConstant;
import com.atguigu.gamll.realtime.utils.ClickHouseUtil;
import com.atguigu.gamll.realtime.utils.DateUtil;
import com.atguigu.gamll.realtime.utils.MyKafkaUtil;
import com.ctc.wstx.util.DataUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

public class ProductStatsApp extends BaseApp {
    public static void main(String[] args) throws Exception {
        ProductStatsApp productStatsApp = new ProductStatsApp();
        productStatsApp.baseEntry();
    }

    @Override
    public void biz(StreamExecutionEnvironment env) {
        // TODO 3.从kafka主题中读取数据
        // 3.1 设置消费主题及消费者组
        String groupId = "product_stats_app";

        String pageViewSourceTopic = "dwd_page_log";
        String orderWideSourceTopic = "dwm_order_wide";
        String paymentWideSourceTopic = "dwm_payment_wide";
        String cartInfoSourceTopic = "dwd_cart_info";
        String favorInfoSourceTopic = "dwd_favor_info";
        String refundInfoSourceTopic = "dwd_order_refund_info";
        String commentInfoSourceTopic = "dwd_comment_info";

        // 3.2 获取消费对象
        FlinkKafkaConsumer<String> pageViewSource  = MyKafkaUtil.getKafkaSource(pageViewSourceTopic,groupId);
        FlinkKafkaConsumer<String> orderWideSource  = MyKafkaUtil.getKafkaSource(orderWideSourceTopic,groupId);
        FlinkKafkaConsumer<String> paymentWideSource  = MyKafkaUtil.getKafkaSource(paymentWideSourceTopic,groupId);
        FlinkKafkaConsumer<String> favorInfoSourceSource  = MyKafkaUtil.getKafkaSource(favorInfoSourceTopic,groupId);
        FlinkKafkaConsumer<String> cartInfoSource  = MyKafkaUtil.getKafkaSource(cartInfoSourceTopic,groupId);
        FlinkKafkaConsumer<String> refundInfoSource  = MyKafkaUtil.getKafkaSource(refundInfoSourceTopic,groupId);
        FlinkKafkaConsumer<String> commentInfoSource  = MyKafkaUtil.getKafkaSource(commentInfoSourceTopic,groupId);

        // 3.3 消费Kafka数据，并封装为流
        DataStreamSource<String> pageViewDS = env.addSource(pageViewSource);
        DataStreamSource<String> favorInfoDS = env.addSource(favorInfoSourceSource);
        DataStreamSource<String> cartInfoDS = env.addSource(cartInfoSource);
        DataStreamSource<String> orderWideDS = env.addSource(orderWideSource);
        DataStreamSource<String> paymentWideDS = env.addSource(paymentWideSource);
        DataStreamSource<String> refundInfoDS = env.addSource(refundInfoSource);
        DataStreamSource<String> commentInfoDS = env.addSource(commentInfoSource);

        // TODO 4.对读取的数据进行类型转换 jsonStr -> ProductStats
        // 点击和曝光
        SingleOutputStreamOperator<ProductStats> clickAndDisplayStatsDS = (SingleOutputStreamOperator<ProductStats>) pageViewDS.process(
                new ProcessFunction<String, ProductStats>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<ProductStats> out) throws Exception {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                        Long ts = jsonObj.getLong("ts");
                        // 判断是否是商品点击时间 pageId = good_detail
                        if ("good_detail".equals(pageJsonObj.getString("page_id"))) {
                            ProductStats productStats = ProductStats.builder()
                                    .sku_id(pageJsonObj.getLong("item"))
                                    .click_ct(1L)
                                    .ts(ts)
                                    .build();
                            out.collect(productStats);
                        }
                        // 判断当前页面上是否有曝光时间
                        JSONArray displaysArr = pageJsonObj.getJSONArray("displays");
                        if (displaysArr != null && displaysArr.size() > 0) {
                            for (int i = 0; i < displaysArr.size(); i++) {
                                JSONObject displaysJsonObj = displaysArr.getJSONObject(i);
                                if ("sku_id".equals(displaysJsonObj.getString("item_type"))) {
                                    ProductStats productStats = ProductStats.builder()
                                            .sku_id(displaysJsonObj.getLong("item"))
                                            .display_ct(1L)
                                            .ts(ts)
                                            .build();
                                    out.collect(productStats);
                                }
                            }
                        }
                    }
                }
        );

        // 收藏
        SingleOutputStreamOperator<ProductStats> favorInfoStatsDS = favorInfoDS.map(
                new MapFunction<String, ProductStats>() {
                    @Override
                    public ProductStats map(String jsonStr) throws Exception {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        return ProductStats.builder()
                                .sku_id(jsonObj.getLong("sku_id"))
                                .ts(DateUtil.toTs(jsonObj.getString("create_time")))
                                .favor_ct(1L)
                                .build();
                    }
                }
        );

        // 加购
        SingleOutputStreamOperator<ProductStats> cartInfoStatsdS = cartInfoDS.map(
                new MapFunction<String, ProductStats>() {
                    @Override
                    public ProductStats map(String jsonStr) throws Exception {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        return ProductStats.builder()
                                .sku_id(jsonObj.getLong("sku_id"))
                                .ts(DateUtil.toTs(jsonObj.getString("create_time")))
                                .cart_ct(1L)
                                .build();
                    }
                }
        );

        // 下单
        SingleOutputStreamOperator<ProductStats> orderWideStatsDS = orderWideDS.map(
                new MapFunction<String, ProductStats>() {
                    @Override
                    public ProductStats map(String jsonStr) throws Exception {
                        OrderWide orderWide = JSON.parseObject(jsonStr, OrderWide.class);
                        return ProductStats.builder()
                                .sku_id(orderWide.getSku_id())
                                .order_sku_num(orderWide.getSku_num())
                                .order_amount(orderWide.getSplit_total_amount())
                                .orderIdSet(new HashSet(Collections.singleton(orderWide.getOrder_id())))
                                .ts(DateUtil.toTs(orderWide.getCreate_time()))
                                .build();
                    }
                }
        );

        // 支付
        SingleOutputStreamOperator<ProductStats> paymentWideStatsDS = paymentWideDS.map(
                new MapFunction<String, ProductStats>() {
                    @Override
                    public ProductStats map(String jsonStr) throws Exception {
                        PaymentWide paymentWide = JSON.parseObject(jsonStr, PaymentWide.class);
                        return ProductStats.builder()
                                .sku_id(paymentWide.getSku_id())
                                .ts(DateUtil.toTs(paymentWide.getCallback_time()))
                                .payment_amount(paymentWide.getSplit_total_amount())
                                .paidOrderIdSet(new HashSet(Collections.singleton(paymentWide.getOrder_id())))
                                .build();
                    }
                }
        );

        // 退单
        SingleOutputStreamOperator<ProductStats> refundInfoStatsDS = refundInfoDS.map(
                new MapFunction<String, ProductStats>() {
                    @Override
                    public ProductStats map(String jsonStr) throws Exception {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        return ProductStats.builder()
                                .sku_id(jsonObj.getLong("sku_id"))
                                .ts(DateUtil.toTs(jsonObj.getString("create_time")))
                                .refund_amount(jsonObj.getBigDecimal("refund_amount"))
                                .refundOrderIdSet(new HashSet(Collections.singleton(jsonObj.getLong("order_id"))))
                                .build();
                    }
                }
        );

        // 评论
        SingleOutputStreamOperator<ProductStats> commentInfoStatsDS = commentInfoDS.map(
                new MapFunction<String, ProductStats>() {
                    @Override
                    public ProductStats map(String jsonStr) throws Exception {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        return ProductStats.builder()
                                .sku_id(jsonObj.getLong("sku_id"))
                                .ts(DateUtil.toTs(jsonObj.getString("create_time")))
                                .comment_ct(1L)
                                .good_comment_ct(
                                        GmallConstant.APPRAISE_GOOD.equals(jsonObj.getString("appraise")) ? 1L : 0L
                                )
                                .build();
                    }
                }
        );

        // TODO 5.union聚合
        DataStream<ProductStats> unionDS = clickAndDisplayStatsDS.union(
                favorInfoStatsDS,
                cartInfoStatsdS,
                orderWideStatsDS,
                paymentWideStatsDS,
                refundInfoStatsDS,
                commentInfoStatsDS
        );

        // TODO 6.设置watermark
        SingleOutputStreamOperator<ProductStats> productStatsWithWatermarkDS = unionDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<ProductStats>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<ProductStats>() {
                            @Override
                            public long extractTimestamp(ProductStats productStats, long l) {
                                return productStats.getTs();
                            }
                        })
        );

        // TODO 7.分组
        KeyedStream<ProductStats, Long> keyedDS = productStatsWithWatermarkDS.keyBy(ProductStats::getSku_id);

        // TODO 8.开窗
        WindowedStream<ProductStats, Long, TimeWindow> windowDS = keyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        // TODO 9.聚合
        SingleOutputStreamOperator<ProductStats> reduceDS = windowDS.reduce(
                new ReduceFunction<ProductStats>() {
                    @Override
                    public ProductStats reduce(ProductStats stats1, ProductStats stats2) throws Exception {
                        stats1.setDisplay_ct(stats1.getDisplay_ct() + stats2.getDisplay_ct());
                        stats1.setClick_ct(stats1.getClick_ct() + stats2.getClick_ct());
                        stats1.setCart_ct(stats1.getCart_ct() + stats2.getCart_ct());
                        stats1.setFavor_ct(stats1.getFavor_ct() + stats2.getFavor_ct());
                        stats1.setOrder_amount(stats1.getOrder_amount().add(stats2.getOrder_amount()));
                        stats1.getOrderIdSet().addAll(stats2.getOrderIdSet());
                        stats1.setOrder_ct(stats1.getOrderIdSet().size() + 0L);
                        stats1.setOrder_sku_num(stats1.getOrder_sku_num() + stats2.getOrder_sku_num());
                        stats1.setPayment_amount(stats1.getPayment_amount().add(stats2.getPayment_amount()));

                        stats1.getRefundOrderIdSet().addAll(stats2.getRefundOrderIdSet());
                        stats1.setRefund_order_ct(stats1.getRefundOrderIdSet().size() + 0L);
                        stats1.setRefund_amount(stats1.getRefund_amount().add(stats2.getRefund_amount()));

                        stats1.getPaidOrderIdSet().addAll(stats2.getPaidOrderIdSet());
                        stats1.setPaid_order_ct(stats1.getPaidOrderIdSet().size() + 0L);

                        stats1.setComment_ct(stats1.getComment_ct() + stats2.getComment_ct());
                        stats1.setGood_comment_ct(stats1.getGood_comment_ct() + stats2.getGood_comment_ct());
                        return stats1;
                    }
                },
                new WindowFunction<ProductStats, ProductStats, Long, TimeWindow>() {
                    @Override
                    public void apply(Long aLong, TimeWindow window, Iterable<ProductStats> input, Collector<ProductStats> out) throws Exception {
                        long stt = window.getStart();
                        long edt = window.getEnd();
                        for (ProductStats productStats : input) {
                            productStats.setStt(DateUtil.toYMDHMS(new Date(stt)));
                            productStats.setEdt(DateUtil.toYMDHMS(new Date(edt)));
                            productStats.setTs(System.currentTimeMillis());
                            out.collect(productStats);
                        }
                    }
                }
        );

        // TODO 10.关联SKU维度
        SingleOutputStreamOperator<ProductStats> productStatsWithSkuDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<ProductStats>("dim_sku_info") {
                    @Override
                    public void join(ProductStats productStats, JSONObject jsonObject) throws Exception {
                        productStats.setSku_name(jsonObject.getString("SKU_NAME"));
                        productStats.setSku_price(jsonObject.getBigDecimal("PRICE"));
                        productStats.setSpu_id(jsonObject.getLong("SPU_ID"));
                        productStats.setCategory3_id(jsonObject.getLong("CATEGORY3_ID"));
                        productStats.setTm_id(jsonObject.getLong("TM_ID"));
                    }

                    @Override
                    public String getKey(ProductStats productStats) {
                        return productStats.getSku_id().toString();
                    }
                },
                60, TimeUnit.SECONDS
        );

        // TODO 11.关联SPU维度
        SingleOutputStreamOperator<ProductStats> productStatsWithSpuDS =
                AsyncDataStream.unorderedWait(
                        productStatsWithSkuDS,
                        new DimAsyncFunction<ProductStats>("DIM_SPU_INFO") {
                            @Override
                            public void join(ProductStats productStats, JSONObject jsonObject) throws Exception {
                                productStats.setSpu_name(jsonObject.getString("SPU_NAME"));
                            }
                            @Override
                            public String getKey(ProductStats productStats) {
                                return String.valueOf(productStats.getSpu_id());
                            }
                        }, 60, TimeUnit.SECONDS);

        // TODO 12.关联类别维度
        SingleOutputStreamOperator<ProductStats> productStatsWithCategory3DS =
                AsyncDataStream.unorderedWait(productStatsWithSpuDS,
                        new DimAsyncFunction<ProductStats>("DIM_BASE_CATEGORY3") {
                            @Override
                            public void join(ProductStats productStats, JSONObject jsonObject) throws Exception {
                                productStats.setCategory3_name(jsonObject.getString("NAME"));
                            }
                            @Override
                            public String getKey(ProductStats productStats) {
                                return String.valueOf(productStats.getCategory3_id());
                            }
                        }, 60, TimeUnit.SECONDS);

        // TODO 13.关联品牌维度
        SingleOutputStreamOperator<ProductStats> productStatsWithTmDS =
                AsyncDataStream.unorderedWait(productStatsWithCategory3DS,
                        new DimAsyncFunction<ProductStats>("DIM_BASE_TRADEMARK") {
                            @Override
                            public void join(ProductStats productStats, JSONObject jsonObject) throws Exception {
                                productStats.setTm_name(jsonObject.getString("TM_NAME"));
                            }
                            @Override
                            public String getKey(ProductStats productStats) {
                                return String.valueOf(productStats.getTm_id());
                            }
                        }, 60, TimeUnit.SECONDS);

        productStatsWithTmDS.print(">>>>>>");

        // TODO 14.写到clickhouse数据库
        productStatsWithTmDS.addSink(
                ClickHouseUtil.<ProductStats>getSinkFunction("insert into product_stats_0819 values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
        );
    }
}
